package com.lagou.common.utils;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Collections;
import java.util.Properties;


public class KafkaUtils {

    /**
     * 私有静态方法
     * 创建Kafka生产者
     * @return
     */
    private static KafkaProducer<String, String> createProducer() {

        //构造一个消息生产者对象,关于kafka集群等相关配置,可以从Properties文件中加载也可以从一个Properties对象中获取
        Properties properties = new Properties();
        //kafka会按照固定的Key取出对应的value
        //制定集群节点
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.30:9092,192.168.200.30:9093,192.168.200" +
                ".30:9094");
        //网络传输,需要对key和value指定对应的序列化类
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        return new KafkaProducer<>((properties));
    }

    /**
     * 私有化静态方法
     * 创建kafka消费者
     * @return
     */
    private static KafkaConsumer<String, String> createConsumer() {

        //构造一个消息生产者对象,关于kafka集群等相关配置,可以从Properties文件中加载也可以从一个Properties对象中获取
        Properties properties = new Properties();
        //kafka会按照固定的Key取出对应的value
        //制定集群节点
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.30:9092,192.168.200.30:9093,192.168" +
                ".200.30:9094");
        //反序列化类
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        //指定组名
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"lagou");
        return new KafkaConsumer<>((properties));
    }

    /**
     * 传入kafka约定的topicName
     * json格式字符串
     * 发送给kafka集群
     * @param topicName
     * @param jsonMessage
     */
    public static void sendMessage(String topicName, String jsonMessage) {
        KafkaProducer<String, String> producer = createProducer();
        producer.send(new ProducerRecord<>(topicName, jsonMessage));
        System.out.println("已经发送消息 : " + jsonMessage);
        producer.close();
    }

    /**
     * 开启kafka消费者消费消息
     * @param topicName
     */
    public static void acceptMessage(String topicName){
        KafkaConsumer<String, String> consumer = createConsumer();
        //订阅消息
        consumer.subscribe(Collections.singletonList(topicName));
        while(true){
            //获取消息是一个阻断式方法
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("主题" + record.topic() + ", 偏移量:" + record.offset() + ", msg:" + record.value());
            }
        }
    }
}
